package cn.doitedu.rtdw.data_etl;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/8
 * @Desc: 学大数据，到多易教育
 * 视频播放行为分析etl任务
 * 任务的核心要点：
 * 1. 为视频播放相关事件数据 关联 视频信息维表
 * 2. 为视频播放相关事件数据增加： start_time字段
 * 3. 进行窗口聚合，求end_time，并依此降低写入doris的数据量和频率
 **/
public class Etl04_VideoPlayActionAnalyse_SQL {
    public static void main(String[] args) {

        // 编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表，来映射kafka中的dwd层的用户行为公共维度宽表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                           "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_province STRING,                     "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     pt AS proctime() ,                            "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 创建hbase连接器表，映射hbase中的视频信息维表
        tenv.executeSql("CREATE TABLE video_hbase( " +
                " video_id STRING, " +
                " f ROW<video_name STRING, video_type STRING, video_album STRING,video_author STRING, video_timelong BIGINT,create_time timestamp(3),update_time timestamp(3) >, " +
                " PRIMARY KEY (video_id) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_video_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");


        // 编写一个lookup join的sql语句，进行维度关联
        tenv.executeSql(
                " CREATE TEMPORARY VIEW play_events AS "
                        + " WITH tmp AS (                                     "
                        + " SELECT                                            "
                        + "   user_id,                                        "
                        + "   release_channel,                                "
                        + "   device_type,                                    "
                        + "   event_id,                                       "
                        + "   properties['play_id']  as play_id,              "
                        + "   properties['video_id'] as video_id,             "
                        + "   event_time,                                     "
                        + "   pt,                                             "
                        + "   rt                                              "
                        + " FROM dwd_kafka                                    "
                        + " WHERE event_id in ('video_play','video_pause',    "
                        + "      'video_hb','video_resume','video_stop')      "
                        + " )                                                 "
                        + "                                                   "
                        + " SELECT                                            "
                        + "   tmp.user_id,                                    "
                        + "   tmp.release_channel,                            "
                        + "   tmp.device_type,                                "
                        + "   tmp.event_id,                                   "
                        + "   tmp.play_id,                                    "
                        + "   tmp.video_id,                                   "
                        + "   tmp.event_time,                                 "
                        + "   tmp.rt,                                         "
                        + "   dim.video_name,                                 "
                        + "   dim.video_type,                                 "
                        + "   dim.video_album,                                "
                        + "   dim.video_author,                               "
                        + "   dim.video_timelong,                             "
                        + "   dim.create_time                                 "
                        + " FROM tmp LEFT JOIN video_hbase                    "
                        + "   FOR SYSTEM_TIME AS OF tmp.pt AS dim             "
                        + "   ON LPAD(tmp.video_id,7,'0') = dim.video_id      "
        );


        tenv.executeSql(
                " CREATE TEMPORARY VIEW play_events_2 AS                                                  "
                +" with tmp AS (                                                                             "
                        +" SELECT                                                                            "
                        +"  user_id,                                                                         "
                        +"  release_channel,                                                                 "
                        +"  device_type,                                                                     "
                        +"  event_id,                                                                        "
                        +"  play_id,                                                                         "
                        +"  video_id,                                                                        "
                        +"  event_time,                                                                      "
                        +"  video_name,                                                                      "
                        +"  video_type,                                                                      "
                        +"  video_album,                                                                     "
                        +"  video_author,                                                                    "
                        +"  video_timelong,                                                                  "
                        +"  rt,                                                                              "
                        +"  if(event_id in ('video_play','video_resume'),1,0) as flag                        "
                        +" from play_events )                                                                "
                        +"                                                                                   "
                        +" SELECT                                                                            "
                        +"   user_id,                                                                        "
                        +"   release_channel,                                                                "
                        +"   device_type,                                                                    "
                        +"   event_id,                                                                       "
                        +"   play_id,                                                                        "
                        +"   video_id,                                                                       "
                        +"   event_time,                                                                     "
                        +"   video_name,                                                                     "
                        +"   video_type,                                                                     "
                        +"   video_album,                                                                    "
                        +"   video_author,                                                                   "
                        +"   video_timelong,                                                                 "
                        +"   rt,                                                                             "
                        +"   first_value(event_time) over(partition by flag order by rt) as start_time       "
                        +" FROM (                                                                            "
                        +"     SELECT                                                                        "
                        +"      user_id,                                                                     "
                        +"      release_channel,                                                             "
                        +"      device_type,                                                                 "
                        +"      event_id,                                                                    "
                        +"      play_id,                                                                     "
                        +"      video_id,                                                                    "
                        +"      event_time,                                                                  "
                        +"      video_name,                                                                  "
                        +"      video_type,                                                                  "
                        +"      video_album,                                                                 "
                        +"      video_author,                                                                "
                        +"      video_timelong,                                                              "
                        +" 	    rt,                                                                          "
                        +"      sum(flag) over(partition by play_id order by rt ) as flag                    "
                        +"     FROM tmp                                                                      "
                        +" ) o                                                                               "
        );

        tenv.executeSql("select * from play_events_2 ").print();

    }
}
